#include "sd_transaction.h"
#include "sd_thread.h"
#include "sd_hashtable.h"
#include "sd_blist.h"
#include "sd_memory.h"
#include "sd_log.h"
#include "sd_ttree.h"


#define sd_TxnGetLockTable(txnState) ((txnState)->txnManager->activeTxns[(txnState)->tid].lockTable)


sd_Ret sd_TxnManagerCreate(sd_TxnManager** txnman, size_t maxNumActive, size_t logListNodeSize, size_t lockHashsize, size_t tableNodeSize,
						   struct timespec txnLockTimeout)
{
	sd_Ret ret;
	uint32 i;

	/* Create the structure of the transaction manager */
	*txnman = (sd_TxnManager*)sd_malloc(sizeof(sd_TxnManager)+(maxNumActive-1)*sizeof(sd_Transaction));
	if (*txnman == NULL)
	{
		sd_log_err("Cannot allocate enough memory to create the transaction manager!\n");
		SD_ASSERT(false);
		return SD_RET_ALLOCATE_MEMEORY;
	}

	/* Create the transaction table for active transactions */
	for (i=0;i<maxNumActive; i++)
	{
		sd_Transaction* txn = &(*txnman)->activeTxns[i];
		txn->isUsed = false;
		txn->tid = i;
		
		/* Create the lock table */
		ret = sd_HashtableCreate(&txn->lockTable, 
			sd_HashFunc_uint64,
			sd_HashKeyCompare_uint64,
			lockHashsize,
			tableNodeSize,
			sizeof(sd_uid),
			sizeof(sd_TxnLockItem),
			NULL);
		SD_ON_FAILED_RETURN(ret);

		/* Create the transaction operation log */
		ret = sd_OpLogCreate(&txn->opLog, logListNodeSize);
		SD_ON_FAILED_RETURN(ret);
	}

	/* Create the record lock table */
	ret = sd_SyncHashtableCreate(&(*txnman)->lockTable,
		sd_HashFunc_uint64,
		sd_HashKeyCompare_uint64,
		lockHashsize,
		tableNodeSize,
		sizeof(sd_uid),
		sizeof(sd_RecordLockItem),
		NULL);
	SD_ON_FAILED_RETURN(ret);

	/* Initialize the variables */
	(*txnman)->maxNumActiveTxns = maxNumActive;
	(*txnman)->freeTxnStartIndex = 0;
	(*txnman)->txnHashsize = lockHashsize;
	(*txnman)->txnHashNodeSize = tableNodeSize;
	(*txnman)->txnLockTimeout = txnLockTimeout;
	
	/* Initialize the lock for operating the transaction manager */
	sd_SpinLock_Init(&(*txnman)->txnPoolLock);

	return SD_RET_OK;
}

sd_Ret sd_TxnManagerFree(sd_TxnManager* txnman)
{
	uint32 i;
	
	if (txnman == NULL)
		return SD_RET_OK;

	sd_SpinLock_Destroy(&txnman->txnPoolLock);

	for (i=0;i<txnman->maxNumActiveTxns; i++)
	{
		sd_Transaction* txn = &txnman->activeTxns[i];
		sd_HashtableFree(txn->lockTable);
		sd_OpLogFree(&txn->opLog);
	}

	/* Free the record lock table */
	sd_SyncHashtableFree(txnman->lockTable);
	
	/* Free the structure */
	sd_free(txnman);
	return SD_RET_OK;
}

sd_Ret sd_TxnManagerBeginTransaction(sd_TxnManager* txnman, sd_TxnState* txnState)
{
	sd_Ret ret;
	uint32 i;
	size_t numScanActiveTxn;
	sd_Transaction* txn;
	bool bFound = false;

	/* Find an active transaction */
	i = txnman->freeTxnStartIndex;
	while (!bFound)
	{
		sd_SpinLock_Lock(&txnman->txnPoolLock);

		numScanActiveTxn = 0;
		while (numScanActiveTxn < txnman->maxNumActiveTxns)
		{
			txn = &txnman->activeTxns[i];
			if (txn->isUsed == false)
			{
				txn->isUsed = true;
				txnState->tid = i;
				txnState->txnManager = txnman;
				txnman->freeTxnStartIndex = i+1;
				if (txnman->freeTxnStartIndex == txnman->maxNumActiveTxns)
					txnman->freeTxnStartIndex = 0;
				ret = SD_RET_OK;
				bFound = true;
				break;
			}
			numScanActiveTxn++;
			i++;
			if (i == txnman->maxNumActiveTxns)
				i = 0;
		}
		if (numScanActiveTxn == txnman->maxNumActiveTxns) 
		{
			/* Not available transaction right now, yield to wait for another transaction */
			sd_SpinLock_Unlock(&txnman->txnPoolLock);
			sched_yield();
		}
		else
		{
			sd_SpinLock_Unlock(&txnman->txnPoolLock);
		}
	}

	return ret;
}

sd_Ret sd_TxnManagerCommitTransaction(sd_TxnState* txnState)
{
	sd_TxnManager* txnman = txnState->txnManager;
	sd_Transaction* txn = &txnman->activeTxns[txnState->tid];

	/* Unlock all the locked nodes in this transaction
	*  This is the second phase in Two Phase Locking Protocol
	*/

	sd_TxnUnlockAll(txnState);

	/* Clear the lock table */
	sd_HashtableClear(txn->lockTable);

	/* Clear the transaction log */
	sd_OpLogClear(&txn->opLog);

	/* Set this transaction is inactive */
	sd_SpinLock_Lock(&txnman->txnPoolLock);
	txn->isUsed = false;
	sd_SpinLock_Unlock(&txnman->txnPoolLock);

	///*for deadlock detection*/
	//sd_SpinLock_Lock(&txnState->txnManager->waitForGraphlock);
	//sd_ReleaseInfoFromWaitForArray(txnState);
	//sd_SpinLock_Unlock(&txnState->txnManager->waitForGraphlock);

	///*for deadlock detection*/

	return SD_RET_OK;
}

sd_Ret sd_TxnManagerAbortTransaction(sd_TxnState* txnState)
{
	sd_Ret ret;
	sd_TxnManager* txnman = txnState->txnManager;
	sd_Transaction* txn = &txnman->activeTxns[txnState->tid];

	/* Rollback all changes recored in transaction log */
	ret = sd_TxnRollback(txnState);
	SD_ON_FAILED_RETURN(ret);

	/* Unlock all the locked nodes in this transaction
	*  This is the second phase in Two Phase Locking Protocol
	*/
	sd_TxnUnlockAll(txnState);

	/* Clear the lock table */
	sd_HashtableClear(txn->lockTable);

	/* Clear the transaction log */
	sd_OpLogClear(&txn->opLog);

	/* Set this transaction is inactive */
	sd_SpinLock_Lock(&txnman->txnPoolLock);
	txn->isUsed = false;
	sd_SpinLock_Unlock(&txnman->txnPoolLock);

	///*for deadlock detection*/
	//sd_SpinLock_Lock(&txnState->txnManager->waitForGraphlock);
	//sd_ReleaseInfoFromWaitForArray(txnState);
	//sd_SpinLock_Unlock(&txnState->txnManager->waitForGraphlock);
	///*for deadlock detection*/

	return SD_RET_OK;
}

sd_Ret sd_TxnRLock(sd_TxnState* txnState, sd_uid rid)
{
	sd_Ret ret;
	sd_Hashtable* table = sd_TxnGetLockTable(txnState);
	sd_SyncHashtableIterator iter;
	sd_TxnLockItem txnItem;
	sd_RecordLockItem recordItem;
	sd_TxnManager* txnMan = txnState->txnManager;

	/* If the transaction has already locked this node, it should release it first */
	if (sd_HashtableLookup(table, &rid, &txnItem) == SD_RET_OK) 
	{
		return SD_RET_OK;
	}
	else
	{
		txnItem.isRLocked = true;
		txnItem.isWLocked = false;

		/* Add this transaction id into record lock table */
		ret = sd_SyncHashtableLookupWithLock(txnMan->lockTable, &rid, &iter, true);
		if (ret == SD_RET_OK)
		{
			/* Get this record item in transaction manager's lock table */
			sd_SyncHashtableGet(&iter, &recordItem);

			(*recordItem.locktimes)++;

			/* Unlock this node in hashtable */
			sd_SyncHashtableReleaseLock(&iter);

			/* Lock this record item's lock */
			if (sd_RWLock_TimeoutRLock(&recordItem.lock, &txnMan->txnLockTimeout) == SD_RET_LOCK_BUSY)
			{
				return SD_RET_DEAD_LOCK;
			}
			

			sd_HashtableInsert(table, &rid, &txnItem);
			return SD_RET_OK;
		}
		else /* This record has not been locked by any transaction before */
		{
			sd_RWLock_Init(&recordItem.lock);

			recordItem.locktimes = (int*)sd_malloc(sizeof(int));
			*recordItem.locktimes = 1;

			sd_SyncHashtableInsertWithLock(&iter, &rid, &recordItem);

			/* Unlock this node in hashtable */
			sd_SyncHashtableReleaseLock(&iter);

			/* Lock this record */
			if (sd_RWLock_TimeoutRLock(&recordItem.lock, &txnMan->txnLockTimeout) == SD_RET_LOCK_BUSY)
			{
				return SD_RET_DEAD_LOCK;
			}

			sd_HashtableInsert(table, &rid, &txnItem);
			return SD_RET_OK;
		}
	}
	return SD_RET_OK;
}

sd_Ret sd_TxnWLock(sd_TxnState* txnState, sd_uid rid)
{
	sd_Ret ret;
	sd_Hashtable* table = sd_TxnGetLockTable(txnState);
	sd_SyncHashtableIterator iter;
	sd_TxnLockItem txnItem;
	sd_RecordLockItem recordItem;
	sd_TxnManager* txnMan = txnState->txnManager;

	/* If the transaction has already locked this node, it should release it first */
	if (sd_HashtableLookup(table, &rid, &txnItem) == SD_RET_OK) 
	{
		if (txnItem.isWLocked == false) /* Only add the Share lock to this record */
		{
			/* Modify this item */
			txnItem.isWLocked = true;
			
			ret = sd_SyncHashtableLookupWithLock(txnMan->lockTable, &rid, &iter, false);	
			/* This record item should exist in lock table, 
			because this transaction has been added in the record item */
			SD_ASSERT(ret == SD_RET_OK);

			/* Get this record item in lock table */
			sd_SyncHashtableGet(&iter, &recordItem);

			/* Unlock this node in hashtable */
			sd_SyncHashtableReleaseLock(&iter);

			/* Lock this record item's lock */
			if (sd_RWLock_TimeoutWLock(&recordItem.lock, &txnMan->txnLockTimeout) == SD_RET_LOCK_BUSY)
			{
				return SD_RET_DEAD_LOCK;
			}

			sd_HashtableUpdate(table, &rid, &txnItem);			
			return SD_RET_OK;
		}
		else /* If this transaction has already Wlocked this record */
		{
			return SD_RET_OK;
		}
	}
	else
	{
		/* Add this record into transaction's lock table */
		txnItem.isRLocked = false;
		txnItem.isWLocked = true;
		/*sd_HashtableInsert(table, &rid, &txnItem);*/ 

		/* Add this transaction id into record lock table */
		ret = sd_SyncHashtableLookupWithLock(txnMan->lockTable, &rid, &iter, true);
		if (ret == SD_RET_OK)
		{
			/* Get this record item in lock table */
			sd_SyncHashtableGet(&iter, &recordItem);
			
			(*recordItem.locktimes)++;

			/* Unlock this node in hashtable */
			sd_SyncHashtableReleaseLock(&iter);

			/* Lock this record */
			if (sd_RWLock_TimeoutWLock(&recordItem.lock, &txnMan->txnLockTimeout) == SD_RET_LOCK_BUSY)
			{
				return SD_RET_DEAD_LOCK;
			}
			
			sd_HashtableInsert(table, &rid, &txnItem);
			return SD_RET_OK;
		}
		else
		{
			/* This hashtable's key and value are both the transaction id (uint32) */
			sd_RWLock_Init(&recordItem.lock);
			recordItem.locktimes = (int*)sd_malloc(sizeof(int));
			*recordItem.locktimes = 1;

			sd_SyncHashtableInsertWithLock(&iter, &rid, &recordItem);

			sd_SyncHashtableReleaseLock(&iter);

			/* Lock this record */
			if (sd_RWLock_TimeoutWLock(&recordItem.lock, &txnMan->txnLockTimeout) == SD_RET_LOCK_BUSY)
			{
				return SD_RET_DEAD_LOCK;
			}

			sd_HashtableInsert(table, &rid, &txnItem);
			return  SD_RET_OK;
		}
		
	}
	return SD_RET_OK;
}

sd_Ret sd_TxnUnlockAll(sd_TxnState* txnState)
{
	sd_Ret ret;
	sd_Hashtable* table = NULL;
	sd_TxnLockItem item;
	sd_HashtableIterator iter;
	sd_SyncHashtableIterator synciter;
	sd_RecordLockItem recordItem;
	sd_uid rid;
	sd_TxnManager* txnMan = txnState->txnManager;

	table = sd_TxnGetLockTable(txnState);
	sd_HashtableBegin(table, &iter);
	while(sd_HashtableNext(&iter) != SD_RET_FALSE)
	{
		sd_HashtableGet(&iter, &item);
		sd_HashtableGetKey(&iter,&rid);
		ret = sd_SyncHashtableLookupWithLock(txnMan->lockTable, &rid, &synciter, false);
		SD_ASSERT(ret == SD_RET_OK);
		sd_SyncHashtableGet(&synciter, &recordItem);
		if (item.isWLocked)
		{
			sd_RWLock_Unlock(&recordItem.lock);
		}
		if (item.isRLocked)
		{
			sd_RWLock_Unlock(&recordItem.lock);
		}
		(*recordItem.locktimes)--;
		if (*recordItem.locktimes == 0)
		{
			sd_free(recordItem.locktimes);
			sd_RWLock_Unlock(&recordItem.lock);
			sd_SyncHashtableRemoveAndReleaseLock(&synciter);
		}
		else
		{
			sd_SyncHashtableReleaseLock(&synciter);
		}
	}
	sd_HashtableClear(table);
	return SD_RET_OK;
}

sd_Ret sd_TxnRollback(sd_TxnState* txnState)
{
	sd_Ret ret;
	sd_OpLog* opLog = sd_TxnGetOpLog(txnState);
	sd_OpLogIterator iter;
	sd_OpLogItem item;
	sd_TTree* ttree;
	sd_Record* record;

	/* Undo each operation in transaction log */
	sd_OpLogBegin(opLog, &iter);
	while(sd_OpLogNext(&iter) != SD_RET_FALSE)
	{
		sd_OpLogGet(&iter, &item);
		
		ttree = item.index;
		record = &item.record;
		
		switch(item.type)
		{
		case SD_OP_INSERT:
			ret = sd_TTreeDelete(ttree, record);
			SD_ON_FAILED_RETURN(ret);
			break;
		case SD_OP_DELETE:
			ret = sd_TTreeInsert(ttree, record);
			SD_ON_FAILED_RETURN(ret);
			break;
		default:
			sd_log_err("Rollback in transaction : Unknown operation in transaction log: %d \n", item.type);
			SD_ASSERT(false);
			break;
		}
	}
	return SD_RET_OK;
}

//sd_Ret sd_AddInfoToWaitForArray(sd_TxnManager* txnMan,sd_Hashtable* txnTable,uint32 tid,bool write)
//{
//    uint32 txnTid;
//	sd_TxnLockItem txnItem;
//	sd_HashtableIterator txnIterator;
//	uint32 maxNumActiveTxns=txnMan->maxNumActiveTxns;
//
//	sd_HashtableBegin(txnTable,&txnIterator);
//	while(sd_HashtableNext(&txnIterator) != SD_RET_FALSE)
//	{
//		sd_HashtableGet(&txnIterator,&txnItem);
//		sd_HashtableGetKey(&txnIterator,&txnTid);
//
//		if(write==false)
//		{
//			if(txnItem.isWLocked !=true)
//		       continue;
//		}
//		/*the information stored in txnWaitForArray is explained as the following:
//		if tid i need the data proposed by tid j, and one of the transaction want to write the data
//		then there is an edge from tid i to tid j in wait-for-graph
//		
//		if in the wait-for-graph, tid i has an edge to tid j, then in the txnWaitForArray
//		    i   j
//		i   0   0
//		j   1   0
//
//		is stored;
//		*/
//		txnMan->txnWaitForArray[txnTid*maxNumActiveTxns+tid]=1;
//
//	}
//    return SD_RET_OK ;
//}
//
//sd_Ret sd_DeleteInfoToWaitForArray(sd_TxnManager* txnMan,sd_Hashtable* txnTable,uint32 tid,bool write)
//{
//
//	uint32 txnTid;
//	sd_TxnLockItem txnItem;
//	sd_HashtableIterator txnIterator;
//	uint32 maxNumActiveTxns=txnMan->maxNumActiveTxns;
//
//	sd_HashtableBegin(txnTable,&txnIterator);
//	while(sd_HashtableNext(&txnIterator) != SD_RET_FALSE)
//	{
//		sd_HashtableGet(&txnIterator,&txnItem);
//		sd_HashtableGetKey(&txnIterator,&txnTid);
//
//		if(write==false)
//		{
//			if(txnItem.isWLocked!=true)
//				continue;
//		}
//		/*the information stored in txnWaitForArray is explained as the following:
//		if tid i need the data proposed by tid j, and one of the transaction want to write the data
//		then there is an edge from tid i to tid j in wait-for-graph
//
//		if in the wait-for-graph, tid i has an edge to tid j, then in the txnWaitForArray
//		i   j
//		i   0   0
//		j   1   0
//
//		is stored;
//		*/
//		txnMan->txnWaitForArray[txnTid*maxNumActiveTxns+tid]=0;
//
//	}
//	
//	return SD_RET_OK ;
//}
//sd_Ret sd_ReleaseInfoFromWaitForArray(sd_TxnState* txnState)
//{
//	uint32 maxNumActiveTxns=txnState->txnManager->maxNumActiveTxns;
//	uint32 *txnWaitForArray=txnState->txnManager->txnWaitForArray;
//	uint32 releaseTid=txnState->tid;
//	uint32 i;
//	
//	for(i=0;i<maxNumActiveTxns;i++)
//	{
//		txnWaitForArray[releaseTid*maxNumActiveTxns+i]=0;
//		txnWaitForArray[i*maxNumActiveTxns+releaseTid]=0;
//	}
//
//	return SD_RET_OK;
//}
//
//sd_Ret sd_TxnCheckLockRing(sd_TxnState* txnState)
//{
//	uint32 checkTid=txnState->tid;
//	uint32 maxNumActiveTxns=txnState->txnManager->maxNumActiveTxns;	
//	uint32 *txnWaitForArray=txnState->txnManager->txnWaitForArray;
//	uint32 *linkedTidsThisBreadth=txnState->txnManager->activeTxns[checkTid].linkedTidsThisBreadth;
//	uint32 *linkedTidsNextBreadth=txnState->txnManager->activeTxns[checkTid].linkedTidsNextBreadth;
//	uint32 i,i_Edges,j_thTxn,k;
//	uint32 buffer,itertionEndBuffer;
//
//	/*tid is between 0 and txnman->maxNumActiveTxns-1*/
//    
//	/*the information stored in txnWaitForArray is explained as the following:
//	if in the wait-for-graph, tid i has an edge to tid j, then in the txnWaitForArray
//	     i   j
//	 i   0   0
//	 j   1   0
//	 
//	 is stored;
//	*/
//
//
//	/*travel breadth-first from the checkTid
//	if checkTid is found in the travel,then there is a ring existing in the wait-for-graph */
//	
//	/*linkedTidsThisBreadth is used to store the tids that  checkTid could arrive with i edges in the wait-for-graph 
//	linkedTidsNextBreadth is used to store the tids that  checkTid could arrive with i+1 edges in the wait-for-graph
//
//	*/
//
//	/*initialize linkedTidsThisBreadth 
//	linkedTidsThisBreadth initialized as   the inverted vector of checkTid*/
//	for(i=0;i<maxNumActiveTxns;i++)
//		linkedTidsThisBreadth[i]=txnWaitForArray[i*maxNumActiveTxns+checkTid];
//
//	
//	/*cal tids__Txn that  checkTid_Txn could arrive with i_Edges in the wait-for-graph */
//	for(i_Edges=2;i_Edges<maxNumActiveTxns;i_Edges++)
//	{
//		/*cal whether checkTid_Txn could arrive  j_thTxn with i_Edges */
//        for(j_thTxn=0;j_thTxn<maxNumActiveTxns;j_thTxn++)
//		{
//			buffer=0;
//
//			for(k=0;k<maxNumActiveTxns;k++)
//			{
//				buffer+=linkedTidsThisBreadth[k]*txnWaitForArray[j_thTxn*maxNumActiveTxns+k];
//			}
//			if(buffer==0)
//				linkedTidsNextBreadth[j_thTxn]=0;
//			else
//				linkedTidsNextBreadth[j_thTxn]=1;
//		
//		}
//		if(linkedTidsNextBreadth[checkTid]==1)
//			return SD_RET_TRUE; 
//		itertionEndBuffer=0;
//		for(i=0;i<maxNumActiveTxns;i++)
//			itertionEndBuffer+=linkedTidsNextBreadth[i];
//
//		if(itertionEndBuffer==0)
//		{
//			/*no ring exist*/
//			break;
//		}
//		for(i=0;i<maxNumActiveTxns;i++)
//			linkedTidsThisBreadth[i]=linkedTidsNextBreadth[i];
//	}
//
//	return SD_RET_FALSE;
//}
